package com.ls.fw.data.hbase.helper;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.ls.fw.commons.core.utils.collections.MapUtils;
import com.ls.fw.commons.core.utils.collections.MapUtils.KeyValueMapper;
import com.ls.fw.data.hbase.bean.Record;
import com.ls.fw.data.hbase.bean.RecordFamily;
import com.ls.fw.data.hbase.support.annotation.HBaseColumn;
import com.ls.fw.data.hbase.utils.DataTypeConvertUtil;

/**
 * Static utility methods for reading (Java) typed fields from an HBase Result.
 */
public class HbaseHelper {

  /**
   * Should never be constructed.
   */
  private HbaseHelper() {
  }

  /**
   * Read the value of this cell and interpret as String.
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static String getString(Result row, String columnFamily, String columnName, String defaultValue) {
    Cell raw = getCell(row,columnFamily,columnName);
    String result = (raw == null) ? defaultValue : Bytes.toString(raw.getValueArray());

    return result;
  }

  public static Cell getCell(Result row, String columnFamily, String columnName){
	  return row.getColumnLatestCell(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
  }
  /**
   * Read the value of this cell and interpret as Integer.
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static Integer getInteger(Result row, String columnFamily, String columnName, Integer defaultValue) {
    Cell raw = getCell(row,columnFamily,columnName);
    Integer result = (raw == null) ? defaultValue : Integer.valueOf(Bytes.toInt(raw.getValueArray()));

    return result;
  }

  /**
   * Read the value of this cell and interpret as Long.
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static Long getLong(Result row, String columnFamily, String columnName, Long defaultValue) {
    Cell raw = getCell(row,columnFamily,columnName);
    Long result = (raw == null) ? defaultValue : Long.valueOf(Bytes.toLong(raw.getValueArray()));

    return result;
  }

  /**
   * Read the value of this cell and interpret as Float.
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static Float getFloat(Result row, String columnFamily, String columnName, Float defaultValue) {
    Cell raw = getCell(row,columnFamily,columnName);
    Float result = (raw == null) ? defaultValue : Float.valueOf(Bytes.toFloat(raw.getValueArray()));

    return result;
  }

  /**
   * Read the value of this cell and interpret as Double.
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static Double getDouble(Result row, String columnFamily, String columnName, Double defaultValue) {
    Cell raw = getCell(row, columnFamily, columnName);
    Double result = (raw == null) ? defaultValue : Double.valueOf(Bytes.toDouble(raw.getValueArray()));

    return result;
  }

  /**
   * Read the value of this cell and return it uninterpreted as byte[].
   *
   * @param row          the HBase Result from which to read
   * @param columnFamily column family that holds the column
   * @param columnName   column or "qualifier"
   * @param defaultValue returned if value at columnName is null
   *
   * @return the value from the specified column, or defaultValue if it's null/doesn't exist
   */
  public static byte[] getBytes(Result row, String columnFamily, String columnName, byte[] defaultValue) {
    Cell raw = getCell(row, columnFamily, columnName);
    byte[] result = (raw == null) ? defaultValue : raw.getValueArray();

    return result;
  }

  public static Long getTimestamp(Result row, String columnFamily, String columnName) {
    Cell raw = getCell(row, columnFamily, columnName);
    return (raw == null) ? null : raw.getTimestamp();
  }
  
  public static Put parse(Record record, ConvertHandler handler) throws Exception{
	  final Put put = new Put(Bytes.toBytes(record.getId()));
	  
	  MapUtils.traversal(record, new KeyValueMapper<String,RecordFamily>(){

		@Override
		public boolean next(String key, RecordFamily value, int index) {
			 final byte[] family = Bytes.toBytes(key);
			 MapUtils.traversal(value, new KeyValueMapper<String,Object>(){

					@Override
					public boolean next(String key, Object value, int index) {
						final byte[] qualifier = Bytes.toBytes(key);
						byte[] val = null;
						if(handler != null){
							val = handler.handler(key, value);
						}else{
							try {
								val = DataTypeConvertUtil.dataConvertByte(value);
							} catch (Exception e) {
								e.printStackTrace();
							}
						}
						put.addColumn(family, qualifier, val);
						return true;
					}
				  });
			return true;
		}
	  });
	  return put;
  }

  public static Record parse(final Result result,String fa, final ColumnHandler columnHandler) throws Exception{
	  final Record r = new Record(Bytes.toString(result.getRow()));
	  NavigableMap<byte[], NavigableMap<byte[], byte[]>>  map = result.getNoVersionMap();
	  MapUtils.traversal(map, new KeyValueMapper<byte[], NavigableMap<byte[], byte[]>>() {
		  	@Override
			public boolean next(byte[] arg0, NavigableMap<byte[], byte[]> arg1,int index){
				String f = Bytes.toString(arg0);
				if(!f.equals(fa)){
					return true;
				}
		  		final RecordFamily family = new RecordFamily(f);
				r.add(family);
				MapUtils.traversal((NavigableMap<byte[], byte[]>)arg1, new KeyValueMapper<byte[], byte[]>() {

					@Override
					public boolean next(byte[] arg0, byte[] arg1,int index) {
						String name = Bytes.toString(arg0);
						boolean handler = true;
						if(columnHandler != null){
							handler = columnHandler.isDeal(name);
						}
						if(handler){
							Object value = Bytes.toString(arg1);
							if(columnHandler != null){
								value = columnHandler.handler(name, arg1);
							}
							if(columnHandler != null){
								name = columnHandler.getName(name);
							}
							family.add(name, value);
						}
						return true;
					}
				});
				return true;
			}
		});
	  return r;
  }
  public static Record parse(final Result result, final ColumnHandler columnHandler) throws Exception{
	  final Record r = new Record(Bytes.toString(result.getRow()));
	  NavigableMap<byte[], NavigableMap<byte[], byte[]>>  map = result.getNoVersionMap();
	  MapUtils.traversal(map, new KeyValueMapper<byte[], NavigableMap<byte[], byte[]>>() {
		  	@Override
			public boolean next(byte[] arg0, NavigableMap<byte[], byte[]> arg1,int index){
				final RecordFamily family = new RecordFamily(Bytes.toString(arg0));
				r.add(family);
				MapUtils.traversal((NavigableMap<byte[], byte[]>)arg1, new KeyValueMapper<byte[], byte[]>() {

					@Override
					public boolean next(byte[] arg0, byte[] arg1,int index) {
						String name = Bytes.toString(arg0);
						boolean handler = true;
						if(columnHandler != null){
							handler = columnHandler.isDeal(name);
						}
						if(handler){
							Object value = Bytes.toString(arg1);
							if(columnHandler != null){
								value = columnHandler.handler(name, arg1);
							}
							if(columnHandler != null){
								name = columnHandler.getName(name);
							}
							family.add(name, value);
						}
						return true;
					}
				});
				return true;
			}
		});
	  return r;
  }

  public static interface ConvertHandler{
	  
	  /**
	   * 返回结果
	   * @param name
	   * @param value
	   * @return
	   */
	  byte[] handler(String name,Object value);
	   
  }

  public static interface ColumnHandler{
	  
	  /**
	   * 返回结果
	   * @param name
	   * @param value
	   * @return
	   */
	  Object handler(String name,byte[] value);
	  
	  /**
	   * 是否处理
	   * @param name
	   * @return
	   */
	  boolean isDeal(String name );
	  
	  /**
	   * 返回名称
	   * @param name
	   * @return
	   */
	  String getName(String name );
  }

  public static <T> T parseToObject(Result result, Class<T> cls)
          throws InstantiationException, IllegalAccessException {
      T resultObject = null;
      resultObject = cls.newInstance();
      Field[] fields = cls.getDeclaredFields();
      for (Cell cell : result.rawCells()) {
          // 列名称
          String rowName = new String(CellUtil.cloneQualifier(cell));
          // 值
          String value = new String(CellUtil.cloneValue(cell));
          for (Field field : fields) {
              HBaseColumn hbaseColumn = field.getAnnotation(HBaseColumn.class);
              // String family = hbaseColumn.family();
              if (null != hbaseColumn) {
                  String qualifier = hbaseColumn.qualifier();
                  if (qualifier.equals(rowName)) {
                      // 打开javabean的访问权限
                      field.setAccessible(true);
                      field.set(resultObject, value);
                  }
              }
          }
      }

      return resultObject;
  }
}
